use actix::prelude::*;
use tokio::net::{TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use std::collections::VecDeque;

// ===== Message for write =====
#[derive(Message)]
#[rtype(result = "()")]
pub struct WriteMsg(pub Vec<u8>);

// ===== TcpActor =====
pub struct TcpActor {
    stream: Option<TcpStream>,
    write_queue: VecDeque<Vec<u8>>,
    is_writing: bool,
    buf: Vec<u8>,
}

impl TcpActor {
    pub fn new(stream: TcpStream) -> Self {
        Self {
            stream: Some(stream),
            write_queue: VecDeque::new(),
            is_writing: false,
            buf: vec![0u8; 1024],
        }
    }

    fn start_reading(&mut self, ctx: &mut Context<Self>) {
        if let Some(mut stream) = self.stream.take() {
            // 读取 buffer 的指针和长度
            let mut buf = vec![0u8; 1024];

            ctx.spawn(
                async move {
                    let n = stream.read(&mut buf).await?;
                    Ok::<(TcpStream, usize, Vec<u8>), std::io::Error>((stream, n, buf))
                }
                .into_actor(self)
                .map(|res, act, ctx| {
                    match res {
                        Ok((stream, n, buf)) => {
                            act.stream = Some(stream); // 放回 stream
                            if n == 0 {
                                println!("Connection closed by peer");
                                ctx.stop();
                            } else {
                                println!("Received: {}", String::from_utf8_lossy(&buf[..n]));
                                act.start_reading(ctx); // 继续读
                            }
                        }
                        Err(e) => {
                            println!("Read error: {}", e);
                            ctx.stop();
                        }
                    }
                }),
            );
        } else {
            println!("No stream to read from");
            ctx.stop();
        }
    }

    fn process_next_write(&mut self, ctx: &mut Context<Self>) {
        if let Some(data) = self.write_queue.pop_front() {
            self.is_writing = true;

            if let Some(mut stream) = self.stream.take() {
                ctx.spawn(
                    async move {
                        stream.write_all(&data).await?;
                        stream.flush().await?;
                        Ok::<TcpStream, std::io::Error>(stream)
                    }
                    .into_actor(self)
                    .map(|res, act, ctx| {
                        match res {
                            Ok(stream) => {
                                act.stream = Some(stream);
                                act.is_writing = false;
                                act.process_next_write(ctx);
                            }
                            Err(e) => {
                                println!("Write error: {}", e);
                                ctx.stop();
                            }
                        }
                    }),
                );
            } else {
                println!("Stream lost before write.");
                ctx.stop();
            }
        } else {
            self.is_writing = false;
        }
    }
}

impl Actor for TcpActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("TcpActor started.");
        self.start_reading(ctx);
    }
}

impl Handler<WriteMsg> for TcpActor {
    type Result = ();

    fn handle(&mut self, msg: WriteMsg, ctx: &mut Context<Self>) {
        self.write_queue.push_back(msg.0);
        if !self.is_writing {
            self.process_next_write(ctx);
        }
    }
}

use actix::prelude::*;
use tokio::net::TcpListener;


#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9000").await?;
    println!("Server listening on 127.0.0.1:9000");

    loop {
        let (stream, addr) = listener.accept().await?;
        println!("New connection from {}", addr);

        let actor = TcpActor::new(stream).start();

        actor.do_send(WriteMsg(b"Hello from Actix TCP!\n".to_vec()));
        actor.do_send(WriteMsg(b"Second message!\n".to_vec()));
    }
}
